package cn.trigram.mq;

import cn.trigram.ddo.entity.Service01OrderEntity;
import cn.trigram.mapper.MqTransactionMapper;
import cn.trigram.mapper.Service01OrderMapper;
import cn.trigram.mapper.dynamic.MqTransactionEntityDynamicSqlSupport;
import java.util.Optional;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.mybatis.dynamic.sql.SqlBuilder;
import org.mybatis.dynamic.sql.render.RenderingStrategies;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.stereotype.Component;

/**
 * 事务监听器中如果要过滤消息只能通过实现不同的RocketMQTemplate 指定不同的topic 和 group
 * @see ExtRocketMQTemplate
 */
@Slf4j
@Setter(onMethod_ = {@Autowired})
@Component
@RocketMQTransactionListener
public class Service01MQLocalTransactionListener implements RocketMQLocalTransactionListener {

  private MqTransactionMapper transactionMapper;

  private Service01OrderMapper orderMapper;

  @Override
  public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {

    log.info("received message: {}", msg);
    MessageHeaders headers       = msg.getHeaders();
    final String   transactionId = headers.get(RocketMQHeaders.TRANSACTION_ID, String.class);
    Object         payload       = msg.getPayload();
    String         uuid          = headers.get("uuid", String.class);
    try {

      Service01OrderEntity orderEntity = new Service01OrderEntity();
      orderEntity.setId(uuid);
      orderEntity.setStatus("finish_order");
      orderMapper.updateByPrimaryKeySelective(orderEntity);
      commit(transactionId);
      return RocketMQLocalTransactionState.COMMIT;
    } catch (Exception e) {

      rollback(transactionId);
//      消息回滚会在rocketmq 中撤销已发送的消息
      return RocketMQLocalTransactionState.ROLLBACK;
    }
  }

  @Override
  public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {

    MessageHeaders headers = msg.getHeaders();
    Object         payload = msg.getPayload();
    final String transactionId = headers.get(
        RocketMQHeaders.TRANSACTION_ID, String.class);
    String                         uuid     = headers.get("uuid", String.class);
    Optional<Service01OrderEntity> optional = orderMapper.selectByPrimaryKey(uuid);
    Service01OrderEntity orderEntity = optional.orElseGet(
        Service01OrderEntity::new);
    if (StringUtils.equals("finish_order", orderEntity.getStatus())) {
      commit(transactionId);
      return RocketMQLocalTransactionState.COMMIT;
    }
    rollback(transactionId);
    return RocketMQLocalTransactionState.ROLLBACK;
  }

  private void commit(String transactionId) {

    transactionMapper.update(
        SqlBuilder.update(MqTransactionEntityDynamicSqlSupport.mqTransactionEntity)
                  .set(
                      MqTransactionEntityDynamicSqlSupport.mqTransactionEntity.status).equalTo(
                      "commit")
                  .where(
                      MqTransactionEntityDynamicSqlSupport.mqTransactionEntity.transactionId,
                      SqlBuilder.isEqualTo(transactionId)
                  ).build().render(RenderingStrategies.MYBATIS3)
    );
  }

  private void rollback(String transactionId) {

    transactionMapper.update(
        SqlBuilder.update(MqTransactionEntityDynamicSqlSupport.mqTransactionEntity)
                  .set(
                      MqTransactionEntityDynamicSqlSupport.mqTransactionEntity.status).equalTo(
                      "rollback")
                  .where(
                      MqTransactionEntityDynamicSqlSupport.mqTransactionEntity.transactionId,
                      SqlBuilder.isEqualTo(transactionId)
                  ).build().render(RenderingStrategies.MYBATIS3)
    );
  }

}